package com.tl.mq.kafka_mq;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;

@RestController
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;
    private final String TOPIC_NAME = "test_topic";

    @PostMapping(value = "/send")
    public void sendMsg(String msg) throws ExecutionException, InterruptedException {
        // 同步阻塞发送
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(this.TOPIC_NAME, msg);
        System.out.println("消息发送： " + future.get().toString());

        // 异步非阻塞发送
        ListenableFuture<SendResult<String, String>> futureSync = kafkaTemplate.send(this.TOPIC_NAME, msg);
        futureSync.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println("消息发送失败！");
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                System.out.println("消息发送成功！！");
            }
        });



    }


}
